package com.xukang.qd.boot;

import com.ibm.mq.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

public class MqHandler {

    public static Logger logger= LoggerFactory.getLogger(MqHandler.class);


    private static String MQ_IP;
    private static int MQ_PORT ;
    // 队列管理器
    private static String MQ_QMANAGER ;
    // mq服务方通道
    private static String MQ_CHANNEL  ;
    // mq通讯编码集
    private static int MQ_CCSID  ;
    // mq请求方发送队列
    private static String MQ_CLENT_SENDQUEUE ;
    // mq请求方接收队列
    private static String MQ_CLENT_RECQUEUE  ;

    // MQ自带连接池
    private static MQSimpleConnectionManager myConnMan = null;

    // 异步长连接接收队列管理器
    private MQQueueManager queueManager = null;

    // 异步发送队列
    private MQQueue SendQueue = null;
    // 异步接收队列
    public MQQueue ReceiveQueue = null;

    /**
     * 构造方法
     */
    public MqHandler() throws Exception {
        init();
    }

    /**
     * 异步发送消息
     *
     * @param msg
     * @param messageId
     * @param charSet
     * @throws Exception
     */
    public void sendMsg(String msg, String messageId, String charSet)
            throws Exception {

        try {

            if (StringUtils.isEmpty(charSet)) {
                charSet = "GBK";
            }

            byte[] msgId = StringUtils.isEmpty(messageId) ? null : messageId
                    .getBytes(charSet);

            // TODO: 2019/8/14   打开队列  ,加了myConnMan为啥打开失败呢？
            queueManager = new MQQueueManager(MQ_QMANAGER, myConnMan);
            SendQueue = queueManager.accessQueue(MQ_CLENT_SENDQUEUE,
                    MQC.MQOO_OUTPUT | MQC.MQPMO_NEW_MSG_ID
                            | MQC.MQOO_FAIL_IF_QUIESCING, null, null, null);

            // 初始化消息选项
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            // 确保每次发送前为消息自动生成唯一的msgId
            pmo.options = pmo.options + MQC.MQPMO_NEW_MSG_ID;
            // 如果设置了该参数，则发送后必须调用commit功能。否则无法将消息发送出
            pmo.options = pmo.options + MQC.MQPMO_SYNCPOINT;

            // 创建消息对象
            MQMessage outMsg = new MQMessage();
            // 设置MQMD格式字段
            outMsg.format = MQC.MQFMT_STRING;

            outMsg.messageId = msgId == null ? MQC.MQMI_NONE : msgId;

            outMsg.encoding = MQ_CCSID;
            outMsg.characterSet = MQ_CCSID;

            // 消息发送时必须以字节流的方式发送
            outMsg.write(msg.getBytes(charSet));

            // 在队列上放置消息
            SendQueue.put(outMsg, pmo);

            // 和MQC.MQPMO_SYNCPOINT属性对应。如果设置了该属性，则发送后需要提交。
            queueManager.commit();

        } finally {
            close(SendQueue);
        }

    }

    /**
     * 接收消息
     * @param correlationId
     * @param charSet
     * @return
     * @throws Exception
     */
    public String receive(String correlationId, String charSet)
            throws Exception {

        try {
            if (StringUtils.isEmpty(charSet)) {
                charSet = "GBK";
            }

            queueManager = new MQQueueManager(MQ_QMANAGER, myConnMan);
            ReceiveQueue = queueManager.accessQueue(MQ_CLENT_RECQUEUE,
                    MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_INQUIRE
                            | MQC.MQOO_FAIL_IF_QUIESCING);

            MQMessage respMessage = new MQMessage();

            // 设置MQMD 格式字段
            respMessage.format = MQC.MQFMT_STRING;
            // 设置编码格式与MQ服务一致
            respMessage.encoding = MQ_CCSID;
            // 设置字符集与MQ服务一致
            respMessage.characterSet = MQ_CCSID;

            MQGetMessageOptions gmo = new MQGetMessageOptions();

            gmo.options = gmo.options + MQC.MQGMO_WAIT; // 如果设置了该参数，则当前线程将阻塞，直到等到回复的消息或超时
            gmo.waitInterval = MQC.MQWI_UNLIMITED;
            if (!StringUtils.isEmpty(correlationId)) {
                // 如果设置了该参数，则根据消息的correlId去匹配对应的响应消息
                gmo.matchOptions = MQC.MQMO_MATCH_CORREL_ID;
                respMessage.correlationId = correlationId.getBytes(charSet);
            }

            ReceiveQueue.get(respMessage, gmo);

            byte[] msgBuffer = new byte[respMessage.getMessageLength()];

            respMessage.readFully(msgBuffer);

            String respMsg = new String(msgBuffer, charSet);

            return respMsg;
        } finally {
            close(ReceiveQueue);
        }
    }

    /**
     * 初始化连接
     *
     * @throws MQException
     */
    private void init() throws MQException {
        MQEnvironment.hostname = MQ_IP;
        MQEnvironment.port = MQ_PORT;
        MQEnvironment.CCSID = MQ_CCSID;
        MQEnvironment.channel = MQ_CHANNEL;
        MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,
                MQC.TRANSPORT_MQSERIES_CLIENT);

        myConnMan = new MQSimpleConnectionManager();
        myConnMan.setActive(MQSimpleConnectionManager.MODE_AUTO);
        myConnMan.setTimeout(3600000);
        myConnMan.setMaxConnections(75);
        myConnMan.setMaxUnusedConnections(50);
        MQEnvironment.setDefaultConnectionManager(myConnMan);

        logger.info("初始化队列管理器receiverQueueManager....." + MQ_QMANAGER);
    }

    private void close(MQQueue queue) {

        try {
            if (queue != null && queue.isOpen) {
                queue.close();
            }

            if (queueManager != null) {
                queueManager.disconnect();
            }

        } catch (MQException ex) {
            logger.error("", ex);
        }
    }

}
